package com.study.flink.source;

import java.io.*;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;

import com.study.flink.entity.TicketsPricing;
import com.study.flink.entity.does.TicketsPricingDO;

import lombok.extern.slf4j.Slf4j;

@Slf4j
public class TicketsPricingSource implements SourceFunction<TicketsPricingDO> {
    private static final long serialVersionUID = 1L;

    private FileInputStream stream;
    private BufferedReader reader;

    private final int watermarkDelayMSecs;
    private final int maxDelayMsecs;
    private final int servingSpeed;

    public TicketsPricingSource() {
        watermarkDelayMSecs = 30000;
        maxDelayMsecs = 10;
        servingSpeed = 1;
    }

    @Override
    public void run(SourceContext<TicketsPricingDO> sourceContext) throws Exception {
        stream = new FileInputStream(TicketsPricing.FIEL_PATH);
        reader = new BufferedReader(new InputStreamReader(stream, "UTF-8"));

        generateLimitedStream(sourceContext);

    }

    private void generateLimitedStream(SourceContext<TicketsPricingDO> sourceContext)
        throws IOException, InterruptedException, ParseException {
        long servingStartTime = Calendar.getInstance().getTimeInMillis();
        long dataStartTime;
        Random rand = new Random(7452);

        PriorityQueue<Tuple2<Long, Object>> emitSchedule = new PriorityQueue<>(32,
            new Comparator<Tuple2<Long, Object>>() {
                @Override
                public int compare(Tuple2<Long, Object> o1, Tuple2<Long, Object> o2) {
                    return o1.f0.compareTo(o2.f0);
                }
            });

        String line;
        TicketsPricingDO ticketsPricingDO;

        // 数据文件中，第一行的标题不要。
        if (reader.ready()) {
            line = reader.readLine();
        }

        if (reader.ready() && (line = reader.readLine()) != null) {
            // read first ride
            ticketsPricingDO = TicketsPricingDO.fromString(line);

            // 提取时间发生的时间戳
            dataStartTime = getEventTime(ticketsPricingDO);

            // 获得延迟时间。
            long delayedEventTime = dataStartTime + getNormalDelayMsecs(rand);

            emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ticketsPricingDO));
            // 安排下一个水印。
            long watermarkTime = dataStartTime + watermarkDelayMSecs;
            Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
            emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
        }
        else {
            return;
        }

        // peek at next ride
        if (reader.ready() && (line = reader.readLine()) != null) {
            ticketsPricingDO = TicketsPricingDO.fromString(line);
        }

        // 一行一行的读取数据文件，并从缓存中随机发送DO对象。
        while (emitSchedule.size() > 0 || reader.ready()) {
            // 将所有事件都插入队列中，这些将会在稍后发射出来。
            long curNextDelayedEventTime = !emitSchedule.isEmpty() ? emitSchedule.peek().f0 : -1;
            long rideEventTime = ticketsPricingDO != null ? getEventTime(ticketsPricingDO) : -1;

            while (ticketsPricingDO != null && (emitSchedule.isEmpty()
                || rideEventTime < curNextDelayedEventTime + maxDelayMsecs)) {

                // 将时间插入到队列
                long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
                emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ticketsPricingDO));

                // 处理下一条数据
                if (reader.ready() && (line = reader.readLine()) != null) {
                    ticketsPricingDO = TicketsPricingDO.fromString(line);
                    rideEventTime = getEventTime(ticketsPricingDO);
                }
                else {
                    ticketsPricingDO = null;
                    rideEventTime = -1;
                }

                log.info("the line: " + line);
            }

            // emit schedule is updated, emit next element in schedule
            Tuple2<Long, Object> head = emitSchedule.poll();
            long delayedEventTime = head.f0;

            long now = Calendar.getInstance().getTimeInMillis();
            // long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
            // long waitTime = servingTime - now;
            long waitTime = 10;

            Thread.sleep((waitTime > 0) ? waitTime : 0);

            if (head.f1 instanceof TicketsPricingDO) {
                TicketsPricingDO emitRide = (TicketsPricingDO) head.f1;
                // emit ride
                sourceContext.collectWithTimestamp(emitRide, getEventTime(emitRide));
            }
            else if (head.f1 instanceof Watermark) {
                Watermark emitWatermark = (Watermark) head.f1;
                // emit watermark
                sourceContext.emitWatermark(emitWatermark);
                // schedule next watermark
                long watermarkTime = delayedEventTime + watermarkDelayMSecs;
                Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
                emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
            }
        }
    }

    @SuppressWarnings("unused")
    private void generateUnorderedStream(SourceContext<TicketsPricingDO> sourceContext)
        throws IOException, InterruptedException, ParseException {
        long servingStartTime = Calendar.getInstance().getTimeInMillis();
        long dataStartTime;
        Random rand = new Random(7452);

        PriorityQueue<Tuple2<Long, Object>> emitSchedule = new PriorityQueue<>(32,
            new Comparator<Tuple2<Long, Object>>() {
                @Override
                public int compare(Tuple2<Long, Object> o1, Tuple2<Long, Object> o2) {
                    return o1.f0.compareTo(o2.f0);
                }
            });

        String line;
        TicketsPricingDO ticketsPricingDO;

        // 数据文件中，第一行的标题不要。
        if (reader.ready()) {
            line = reader.readLine();
        }

        if (reader.ready() && (line = reader.readLine()) != null) {
            // read first ride
            ticketsPricingDO = TicketsPricingDO.fromString(line);

            // 提取时间发生的时间戳
            dataStartTime = getEventTime(ticketsPricingDO);

            // 获得延迟时间。
            long delayedEventTime = dataStartTime + getNormalDelayMsecs(rand);

            emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ticketsPricingDO));
            // 安排下一个水印。
            long watermarkTime = dataStartTime + watermarkDelayMSecs;
            Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
            emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
        }
        else {
            return;
        }

        // peek at next ride
        if (reader.ready() && (line = reader.readLine()) != null) {
            ticketsPricingDO = TicketsPricingDO.fromString(line);
        }

        // 一行一行的读取数据文件，并从缓存中随机发送DO对象。
        while (emitSchedule.size() > 0 || reader.ready()) {
            // 将所有事件都插入队列中，这些将会在稍后发射出来。
            long curNextDelayedEventTime = !emitSchedule.isEmpty() ? emitSchedule.peek().f0 : -1;
            long rideEventTime = ticketsPricingDO != null ? getEventTime(ticketsPricingDO) : -1;

            while (ticketsPricingDO != null && (emitSchedule.isEmpty()
                || rideEventTime < curNextDelayedEventTime + maxDelayMsecs)) {

                // 将时间插入到队列
                long delayedEventTime = rideEventTime + getNormalDelayMsecs(rand);
                emitSchedule.add(new Tuple2<Long, Object>(delayedEventTime, ticketsPricingDO));

                // 处理下一条数据
                if (reader.ready() && (line = reader.readLine()) != null) {
                    ticketsPricingDO = TicketsPricingDO.fromString(line);
                    rideEventTime = getEventTime(ticketsPricingDO);
                }
                else {
                    ticketsPricingDO = null;
                    rideEventTime = -1;
                }

                log.info("the line: " + line);
            }

            // emit schedule is updated, emit next element in schedule
            Tuple2<Long, Object> head = emitSchedule.poll();
            long delayedEventTime = head.f0;

            long now = Calendar.getInstance().getTimeInMillis();
            long servingTime = toServingTime(servingStartTime, dataStartTime, delayedEventTime);
            long waitTime = servingTime - now;

            Thread.sleep((waitTime > 0) ? waitTime : 0);

            if (head.f1 instanceof TicketsPricingDO) {
                TicketsPricingDO emitRide = (TicketsPricingDO) head.f1;
                // emit ride
                sourceContext.collectWithTimestamp(emitRide, getEventTime(emitRide));
            }
            else if (head.f1 instanceof Watermark) {
                Watermark emitWatermark = (Watermark) head.f1;
                // emit watermark
                sourceContext.emitWatermark(emitWatermark);
                // schedule next watermark
                long watermarkTime = delayedEventTime + watermarkDelayMSecs;
                Watermark nextWatermark = new Watermark(watermarkTime - maxDelayMsecs - 1);
                emitSchedule.add(new Tuple2<Long, Object>(watermarkTime, nextWatermark));
            }
        }
    }

    public long toServingTime(long servingStartTime, long dataStartTime, long eventTime) {
        long dataDiff = eventTime - dataStartTime;
        return servingStartTime + (dataDiff / this.servingSpeed);
    }

    /**
     * 获得时间发生事件
     * 
     * @author yandong
     * @description 这里我们认为 insertDate 便是事件发生事件。
     * @date 2019年7月12日 下午7:08:28
     * @version
     * @param ticketsPricingDO
     * @return
     * @throws ParseException
     */
    private long getEventTime(TicketsPricingDO ticketsPricingDO) throws ParseException {
        return new SimpleDateFormat("YY-MM-dd HH:mm:ss").parse(ticketsPricingDO.getInsertDate())
            .getTime();
    }

    /**
     * 获得延迟的标准时长（毫秒）
     * 
     * @author yandong
     * @description
     * @date 2019年7月12日 下午7:03:44
     * @version
     * @param rand
     * @return
     */
    private long getNormalDelayMsecs(Random rand) {
        return 100;
    }

    @Override
    public void cancel() {
    }

}
